package com.guiji.job.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.guiji.common.core.domain.AjaxResult;
import com.guiji.common.core.domain.model.LoginUser;
import com.guiji.common.utils.JsonUtils;
import com.guiji.common.utils.PageUtils;
import com.guiji.common.core.page.PagePlus;
import com.guiji.common.core.page.TableDataInfo;
import com.guiji.common.utils.SecurityUtils;
import com.guiji.job.vo.UnTagListVo;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.GetAliasesResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.cluster.metadata.AliasMetadata;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import com.guiji.common.core.mybatisplus.core.ServicePlusImpl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.guiji.job.bo.JobAddBo;
import com.guiji.job.bo.JobQueryBo;
import com.guiji.job.bo.JobEditBo;
import com.guiji.job.domain.Job;
import com.guiji.job.mapper.JobMapper;
import com.guiji.job.vo.JobVo;
import com.guiji.job.service.IJobService;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

/**
 * 任务主Service业务层处理
 *
 * @author 钟维澎（四月）
 * @date 2021-07-02
 */
@Service
public class JobServiceImpl extends ServicePlusImpl<JobMapper, Job> implements IJobService {

	private static final Logger logger = LoggerFactory.getLogger(JobServiceImpl.class);

	@Qualifier("restHighLevelClient")
	@Autowired
	private RestHighLevelClient client;
	@Autowired
	JobMapper jobMapper;

	@Override
	public JobVo queryById(Long id) {
		return getVoById(id, JobVo.class);
	}

	@Override
	public TableDataInfo<UnTagListVo> queryPendingList(JobQueryBo bo) {
		LoginUser loginUser = SecurityUtils.getLoginUser();
		String userName = loginUser.getUser().getUserName();
		bo.setPerson(userName);
		bo.setStatus(2);
		PagePlus<Object, Object> objectObjectPagePlus = PageUtils.buildPagePlus();
		IPage<UnTagListVo> list = jobMapper.getUnTagListVo(objectObjectPagePlus, bo);
		return PageUtils.buildDataInfo(list);
	}


	@Override
	public TableDataInfo<JobVo> queryPageList(JobQueryBo bo) {
		PagePlus<Job, JobVo> result = new PagePlus<>();
		IPage<JobVo> jobVoIPage = baseMapper.selectJobPageVo(PageUtils.buildPagePlus(), bo);
		List<JobVo> volist = jobVoIPage.getRecords().stream()
			.map(any -> BeanUtil.toBean(any, JobVo.class))
			.collect(Collectors.toList());
		result.setRecordsVo(volist);
		result.setTotal(jobVoIPage.getTotal());
		return PageUtils.buildDataInfo(result);
	}

	@Override
	public List<JobVo> queryList(JobQueryBo bo) {
		return listVo(buildQueryWrapper(bo), JobVo.class);
	}

	private LambdaQueryWrapper<Job> buildQueryWrapper(JobQueryBo bo) {
		Map<String, Object> params = bo.getParams();
		LambdaQueryWrapper<Job> lqw = Wrappers.lambdaQuery();
		lqw.eq(bo.getId() != null, Job::getId, bo.getId());
		lqw.eq(bo.getResourceId() != null, Job::getResourceId, bo.getResourceId());
		lqw.eq(bo.getTemplateId() != null, Job::getTemplateId, bo.getTemplateId());
		lqw.eq(bo.getStatus() != null, Job::getStatus, bo.getStatus());
		lqw.eq(StrUtil.isNotBlank(bo.getPerson()), Job::getPerson, bo.getPerson());
		lqw.eq(bo.getVersion() != null, Job::getVersion, bo.getVersion());
		lqw.eq(bo.getCreateTime() != null, Job::getCreateTime, bo.getCreateTime());
		lqw.eq(StrUtil.isNotBlank(bo.getCreateBy()), Job::getCreateBy, bo.getCreateBy());
		lqw.eq(bo.getUpdateTime() != null, Job::getUpdateTime, bo.getUpdateTime());
		lqw.eq(StrUtil.isNotBlank(bo.getUpdateBy()), Job::getUpdateBy, bo.getUpdateBy());
		return lqw;
	}

	@Override
	public Boolean insertByAddBo(JobAddBo bo) {
		Job add = BeanUtil.toBean(bo, Job.class);
		validEntityBeforeSave(add);
		return save(add);
	}

	@Override
	public Boolean updateByEditBo(JobEditBo bo) {
		Job update = BeanUtil.toBean(bo, Job.class);
		if (bo.getIds() != null && bo.getIds().length > 0) {
			for (Long id : bo.getIds()) {
				Long version = getById(id).getVersion();
				update.setId(id);
				update.setVersion(version);
				updateById(update);
			}
			return true;
		}
		validEntityBeforeSave(update);
		return updateById(update);
	}

	/**
	 * 保存前的数据校验
	 *
	 * @param entity 实体类数据
	 */
	private void validEntityBeforeSave(Job entity) {
		//TODO 做一些数据校验,如唯一约束
	}

	@Override
	public Boolean deleteWithValidByIds(Collection<Long> ids, Boolean isValid) {
		if (isValid) {
			//TODO 做一些业务上的校验,判断是否需要校验
		}
		return removeByIds(ids);
	}

	@Override
	public TableDataInfo<UnTagListVo> queryUncheckPageList(JobQueryBo bo) {
		/*LoginUser loginUser = SecurityUtils.getLoginUser();
		String userName = loginUser.getUser().getUserName();
		bo.setPerson(userName);*/
		bo.setStatus(3);
		PagePlus<Object, Object> objectObjectPagePlus = PageUtils.buildPagePlus();
		IPage<UnTagListVo> list = jobMapper.getUnTagListVo(objectObjectPagePlus, bo);
		return PageUtils.buildDataInfo(list);
	}

	@Override
	public Set<String> getEsIndex() {
		GetAliasesRequest request = new GetAliasesRequest();
		GetAliasesResponse getAliasesResponse = null;
		try {
			getAliasesResponse = client.indices().getAlias(request, RequestOptions.DEFAULT);
		} catch (IOException e) {
			e.printStackTrace();
		}
		Map<String, Set<AliasMetadata>> aliases = getAliasesResponse.getAliases();
		Set<String> indices = new HashSet<>();
		for (String s : aliases.keySet()) {
			if (!s.startsWith(".")) {
				indices.add(s);
			}
		}
		return indices;
	}

	@Override
	public AjaxResult bulkInsertEs(JobQueryBo bo) {
		if (ObjectUtil.isEmpty(bo.getTemplateId())) {
			return AjaxResult.error("template_id不能为空");
		}
		if (ObjectUtil.isEmpty(bo.getIndexName())) {
			return AjaxResult.error("indexName不能为空");
		}
		List<Map<String, String>> list = jobMapper.selectEsJson(bo);
		if (ObjectUtil.isEmpty(list)) {
			return AjaxResult.error("没有可入库的数据");
		}

		List<Map<String, String>> newList = new ArrayList<>();
		SimpleDateFormat sdf = new SimpleDateFormat("MM-dd");
		String index = bo.getIndexName() + "-" + sdf.format(new Date());
		BulkProcessor bulkProcessor = getBulkProcessor(client);
		try {
			for (Map<String, String> map : list) {
				HashMap hashMap = JsonUtils.parseObject(map.get("backup_1"), HashMap.class);
				hashMap.put("upload_path", map.get("upload_path"));
				newList.add(hashMap);

				int size = list.size();
				if (size % 1000 == 0) {
					logger.info("mysql handle data  number:" + size);
					// 将数据添加到 bulkProcessor
					list.forEach(hashMap2 -> {
						bulkProcessor.add(new IndexRequest(index).source(hashMap2));
					});
					map.clear();
					list.clear();
				}
			}
			// 处理 未提交的数据
			list.forEach(hashMap3 -> {
				bulkProcessor.add(new IndexRequest(index).source(hashMap3));
			});
			bulkProcessor.flush();
		} catch (Exception e) {
			e.printStackTrace();
			return AjaxResult.error("入库失败" + e.getMessage());
		}finally {
			try {
				boolean terminatedFlag = bulkProcessor.awaitClose(150L, TimeUnit.SECONDS);
				logger.info(Convert.toStr(terminatedFlag));
			} catch (Exception e) {
				logger.error(e.getMessage());
			}
		}
		return AjaxResult.success();
	}


	private BulkProcessor getBulkProcessor(RestHighLevelClient client) {

		BulkProcessor bulkProcessor = null;
		try {

			BulkProcessor.Listener listener = new BulkProcessor.Listener() {
				@Override
				public void beforeBulk(long executionId, BulkRequest request) {
					logger.info("Try to insert data number : " + request.numberOfActions());
				}

				@Override
				public void afterBulk(long executionId, BulkRequest request,
									  BulkResponse response) {
					logger.info("************** Success insert data number : " + request.numberOfActions() + " , id: " + executionId);
				}

				@Override
				public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
					logger.error("Bulk is unsuccess : " + failure + ", executionId: " + executionId);
				}
			};

			BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer = (request, bulkListener) -> client
				.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);

			BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);
			builder.setBulkActions(5000);
			builder.setBulkSize(new ByteSizeValue(100L, ByteSizeUnit.MB));
			builder.setConcurrentRequests(10);
			builder.setFlushInterval(TimeValue.timeValueSeconds(100L));
			builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));
			// 注意点：让参数设置生效
			bulkProcessor = builder.build();

		} catch (Exception e) {
			e.printStackTrace();
			try {
				bulkProcessor.awaitClose(100L, TimeUnit.SECONDS);
			} catch (Exception e1) {
				logger.error(e1.getMessage());
			}
		}
		return bulkProcessor;
	}


	@Override
	public TableDataInfo<Map> countStatus() {
		PagePlus<Object, Object> objectObjectPagePlus = PageUtils.buildPagePlus();
		IPage<Map> list = jobMapper.countStatus(objectObjectPagePlus);
		return PageUtils.buildDataInfo(list);
	}

}
